package org.tao.lightningmq.nameserver.replication;

import com.alibaba.fastjson2.JSON;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import org.tao.lightningmq.common.dto.SlaveAckDTO;
import org.tao.lightningmq.common.enums.NameServerEventCode;
import org.tao.lightningmq.common.enums.NameServerResponseCode;
import org.tao.lightningmq.common.tcp.TcpMsg;
import org.tao.lightningmq.nameserver.common.CommonCache;
import org.tao.lightningmq.nameserver.common.MasterSlaveReplicationProperties;
import org.tao.lightningmq.nameserver.enums.MasterSlaveReplicationTypeEnum;
import org.tao.lightningmq.nameserver.event.model.ReplicationMsgEvent;

import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Author lht
 * @date 2025/1/17 - 16:19
 * @description: 主从之间同步数据发送任务
 */
public class MasterReplicationMsgSendTask extends BaseReplicationTask {

    public MasterReplicationMsgSendTask(String taskName) {
        super(taskName);
    }

    @Override
    protected void startTask() {
        MasterSlaveReplicationProperties masterSlaveReplicationProperties = CommonCache.getNameserverProperties().getMasterSlaveReplicationProperties();
        MasterSlaveReplicationTypeEnum replicationTypeEnum = MasterSlaveReplicationTypeEnum.getEnumByCode(masterSlaveReplicationProperties.getType());
        while (true) {
            try {
                ReplicationMsgEvent replicationMsgEvent = CommonCache.getReplicationMsgQueueManager().getReplicationMsgEvents().take();
                Channel brokerChannel = replicationMsgEvent.getContext().channel();
                Map<String, ChannelHandlerContext> validSlaveChannelMap = CommonCache.getReplicationChannelManager().getValidSlaveChannelMap();
                int validSlaveChannelCount = validSlaveChannelMap.keySet().size();
                switch (replicationTypeEnum) {
                    // 异步复制，直接发送同步数据，同时返回注册成功信号给到broker节点
                    case ASYNC:
                        brokerChannel.writeAndFlush(new TcpMsg(NameServerResponseCode.REGISTRY_SUCCESS));
                        break;
                    // 同步复制，发送同步数据给到slave节点，slave节点返回ack信号，主节点收到ack信号后通知给broker注册成功
                    case SYNC:
                        this.inputMsgToAckMap(replicationMsgEvent, validSlaveChannelCount);
                        break;
                    // 半同步复制，只需要接收到一半的ack次数
                    case HALF_SYNC:
                        this.inputMsgToAckMap(replicationMsgEvent, validSlaveChannelCount / 2);
                        break;
                    default:
                        throw new RuntimeException("未知的复制模式");
                }
                this.sendMsgToSlave(replicationMsgEvent);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
    }

    /**
     * 发送消息给到slave节点
     * @param replicationMsgEvent 复制消息事件
     */
    private void sendMsgToSlave(ReplicationMsgEvent replicationMsgEvent) {
        replicationMsgEvent.setContext(null);
        TcpMsg tcpMsg = new TcpMsg(
                NameServerEventCode.MASTER_REPLICATION_MSG.getCode(), JSON.toJSONBytes(replicationMsgEvent));
        CommonCache.getReplicationChannelManager().getValidSlaveChannelMap().forEach((k, v) -> {
            v.writeAndFlush(tcpMsg);
        });
    }

    /**
     * 将主节点需要发送出去的数据注入到一个map中，然后当从节点返回ack的时候，该map的数据会被剔除对应记录
     * @param replicationMsgEvent
     * @param needAckCount
     */
    private void inputMsgToAckMap(ReplicationMsgEvent replicationMsgEvent, int needAckCount) {
        CommonCache.getAckMap().put(
                replicationMsgEvent.getMsgId(),
                new SlaveAckDTO(new AtomicInteger(needAckCount),
                        replicationMsgEvent.getContext())
        );
    }

}
